package com.atguigu.flink.watermark;

import com.alibaba.fastjson.JSON;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.file.src.FileSource;
import org.apache.flink.connector.file.src.reader.TextLineInputFormat;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * Created by Smexy on 2023/3/31
 *
        老的api，无法从源头产生水印，只能读完数据，手动调用 .assignTimestampsAndWatermarks(watermarkStrategy)
        产生水印。

        新的api，fromSource(xx,WatermarkStrategy x,xx),可以直接基于这个水印策略从源头产生水印，无需调用assignTimestampsAndWatermarks
        算子。
 *
 */
public class Demo9_SourceGenerateWaterMark
{
    public static void main(String[] args) throws Exception {

        Configuration conf = new Configuration();
        conf.setInteger("rest.port",3333);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);

        FileSource<String> source = FileSource.forRecordStreamFormat(
            new TextLineInputFormat(),
            new Path("data/sensor.json")
        ).build();

        WatermarkStrategy<String> watermarkStrategy = WatermarkStrategy
            .<String>forMonotonousTimestamps()
            //从json的每行数据中提取 ts属性值作为 时间属性
            .withTimestampAssigner( (e, ts) -> JSON.parseObject(e).getLong("ts") );

        env.fromSource(source, watermarkStrategy,"myfile")
           .print();

        env.execute();

    }
}
